package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"time"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	// args := ExampleArgs{}

	// declare a reply structure.
	// reply := ExampleReply{}

	for {
		args := ExampleArgs{X: 105}
		//CallExample()
		// declare a reply structure.
		reply := ExampleReply{Y: 100}
		// if call("Coordinator.Done", &args, &reply) && reply.done {
		// 	break
		// }
		if call("Coordinator.Request", &args, &reply) {
			//fmt.Printf("reply.woker : %s %s \n", reply.Woker, reply.Files)

			switch reply.Woker {
			case "map":
				kv := []KeyValue{}
				for _, filename := range reply.Files {
					//fmt.Printf("%s", filename)
					file, err := os.Open(filename)
					if err != nil {
						log.Fatalf("cannot open %v", filename)
					}
					content, err := ioutil.ReadAll(file)
					if err != nil {
						log.Fatalf("cannot read %v", filename)
					}
					// content, err := ioutil.ReadAll(v)
					kv = append(kv, mapf(filename, string(content))...)
				}

				nred := reply.NReduce
				id := reply.Id
				keys := make([][]KeyValue, nred)

				for _, v := range kv {
					idx := ihash(v.Key) % nred
					//fmt.Println("id:" + fmt.Sprintf("%d", idx))
					keys[idx] = append(keys[idx], v)
				}

				files := make([]string, nred)

				for i, v := range keys {
					files[i] = "mr-tmp-" + strconv.Itoa(reply.Id) + "-" + strconv.Itoa(i)
					ofile, _ := os.Create(files[i])
					enc := json.NewEncoder(ofile)
					err := enc.Encode(v)
					if err != nil {
						return 
					}
					err = ofile.Close()
					if err != nil {
						return 
					}
				}

				args = ExampleArgs{Id: id, Worker: "map", Files: files, EndTime: reply.EndTime}

			case "reduce":
				file := reply.Files
				intermediate := ByKey{}
				for _, f := range file {
					open, err := os.Open(f)
					if err != nil {
						return
					}
					dec := json.NewDecoder(open)
					//var kv KeyValue
					inte := ByKey{}
					if err := dec.Decode(&inte); err != nil {
						log.Fatalf("cannot read %v", err)
						//break
					}
					intermediate = append(intermediate, inte...)
				}


				sort.Sort(intermediate)
				
				filename := fmt.Sprintf("mr-out-%d", reply.Id)
				//ofile := fmt.Sprintf"mr-out-%d", reply.Id)
				ofile, _ := os.Create(filename)
				i := 0
				for i < len(intermediate) {
					j := i + 1
					for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
						j++
					}
					var values []string
					for k := i; k < j; k++ {
						values = append(values, intermediate[k].Value)
					}
					output := reducef(intermediate[i].Key, values)

					// this is the correct format for each line of Reduce output.
					//fmt.Printf( "%v %v\n", intermediate[i].Key, output)
					_, err := fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)
					if err != nil {
						return
					}

					i = j
				}
				err := ofile.Close()
				if err != nil {
					return
				}
				time.Sleep(time.Second)

				args = ExampleArgs{Id: reply.Id, Worker: "reduce", EndTime: reply.EndTime}
			default:
				time.Sleep(time.Second)
				continue
			}

			call("Coordinator.Commit", &args, nil)
		} else {
			// sleep
			break
		}
	}

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()
}

func mapcall(ags *ExampleArgs) {
	// call("Coordinator.Commit", &args, &reply)
}

func reducecall(ags *ExampleArgs) {

}

//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Coordinator.Example", &args, &reply)

	// reply.Y should be 100.
	//fmt.Printf("reply.Y %v\n", reply.Y)
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	marshal, _ := json.Marshal(args)
	fmt.Printf("rpcname:%v args:%v \n", rpcname, string(marshal))

	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		//log.Fatal("dialing:", err)
		return false
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
